package com.zhaotianhao.kafka;

import com.alibaba.fastjson.JSON;
import com.github.pagehelper.PageInfo;
import com.zhaotianhao.mapper.MoneyFlowMapper;
import com.zhaotianhao.pojo.Goods;
import com.zhaotianhao.pojo.MoneyFlow;
import com.zhaotianhao.pojo.Order;
import com.zhaotianhao.service.GoodsService;
import com.zhaotianhao.service.OrderService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @Author Z-7
 * @Date 2022/3/7
 */
public class Kafka implements AcknowledgingMessageListener<String, String> {
    @Autowired
    private OrderService orderService;
    @Autowired
    private GoodsService goodsService;
    @Autowired
    private MoneyFlowMapper moneyFlowMapper;
    @Resource
    private RedisTemplate redisTemplate;
    @Override
    public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
        if(data.key().equals("order")){
            System.out.println("kafka 开始消费消息");
            Order order = JSON.parseObject(data.value(), Order.class);
            // 防止重复消费
            if(!(redisTemplate.opsForValue().setIfAbsent(order.getId(),"",1, TimeUnit.HOURS))){
                System.out.println("重复消费，退出");
                return;
            }
            orderService.updateOrderStatus(order.getId(), 1);
            MoneyFlow moneyFlow = new MoneyFlow();
            moneyFlow.setMoney(order.getFactPrice()).setOrderId(order.getId()).setPayType(1);
            moneyFlowMapper.insert(moneyFlow);
            System.out.println("消费完成");
        }else if(data.key().equals("edit")){
            if(!(redisTemplate.opsForValue().setIfAbsent(data.value(),"",1, TimeUnit.HOURS))){
                System.out.println("商品编辑：重复消费，退出");
                return;
            }
            // 更新redis缓存
            PageInfo<Goods> manage = goodsService.queryByPage(null, 1, 3);
            redisTemplate.opsForValue().set("manage",manage,10, TimeUnit.MINUTES);
            System.out.println(manage);
            System.out.println("goods首页更新完毕");
        }

        // 手工确认提交
        acknowledgment.acknowledge();
    }
}
